import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Demo02_transformation {

    /**
     * 简单转换算子 DataStream转成DataStream
     * @param oriDataStream
     */
    public static DataStream<Demo02_sensorReading> transformation_simple(DataStreamSource<String> oriDataStream){

        DataStream<Demo02_sensorReading> sensorDataStream = oriDataStream.map(
                value -> {
                    String[] fields = value.split(",");
                    return new Demo02_sensorReading(
                            fields[0],
                            Long.parseLong(fields[1]),
                            Double.parseDouble(fields[2])
                    );
                }
        );

        return sensorDataStream;
//        sensorDataStream.print();

    }

    /**
     * 键控算子+滚动聚合算子
     * 键控算子是把DataStream转成KeyedStream
     * 滚动聚合算子把KeyedStream转成DataStream
     * @param sensorDataStream
     */
    public static void transformation_keyBy(DataStream<Demo02_sensorReading> sensorDataStream){
        //1.使用keyBy算子，确定key是谁
        KeyedStream<Demo02_sensorReading,String> keyedStream = sensorDataStream.keyBy(Demo02_sensorReading::getSensorId);

        //min只输出当前字段的最小值，其他字段使用第一条数据的内容
        keyedStream.min("temperature").print("min");
        //minBy输出当前字段的最小值的那一整条数据，但是时间戳不是最新的
        keyedStream.minBy("temperature").print("minBy");

        //可以根据实际情况确定返回的内容
        keyedStream.reduce(
            (value1,value2)->
                new Demo02_sensorReading(
                        value1.getSensorId(),
                        value2.getTimestamp(),
                        Math.min(value1.getTemperature(), value2.getTemperature())
                )
        ).print("reduce");
    }

    /**
     * 分流操作
     * 必须使用底层API,以侧输出流的形式进行分流
     * @param sensorDataStream
     */
    public static void transformation_split(DataStream<Demo02_sensorReading> sensorDataStream){

        SingleOutputStreamOperator<Double> tempDataStream = sensorDataStream.process(
                new ProcessFunction<Demo02_sensorReading, Double>() {
                    @Override
                    public void processElement(Demo02_sensorReading value, ProcessFunction<Demo02_sensorReading, Double>.Context ctx, Collector<Double> out) throws Exception {
                        double temp = value.getTemperature();

                        //如果温度值小于等于10或者温度值大于等于30 分别以两条侧输出流进行输出，大于10小于30的数据以正常流的形式输出
                        if(temp<=10){
                            ctx.output(new OutputTag<Double>("cold"){},temp);
                        }else if(temp>=30){
                            ctx.output(new OutputTag<Double>("hot"){},temp);
                        }else{
                            out.collect(temp);
                        }
                    }
                }
        );

        tempDataStream.print("comfortable");
        tempDataStream.getSideOutput(new OutputTag<Double>("cold"){}).print("cold");
        tempDataStream.getSideOutput(new OutputTag<Double>("hot"){}).print("hot");
    }

    /**
     * 合并相同类型的流
     * @param sensorDataStream1
     * @param sensorDataStream2
     */
    public static void transformation_union(DataStream<Demo02_sensorReading> sensorDataStream1,DataStream<Demo02_sensorReading> sensorDataStream2){
        DataStream<Demo02_sensorReading> sensorDataStreamAll = sensorDataStream1.union(sensorDataStream2);

        sensorDataStreamAll.print();
    }

    /**
     * 合并不同类型的流
     * connect方法后转成ConnectedStreams，ConnectedStreams调用方法转成 DataStream
     * @param sensorDataStream1
     * @param sensorDataStream2
     */
    public static void transformation_connect(DataStream<Demo02_sensorReading> sensorDataStream1,DataStream<Demo02_sensorReading> sensorDataStream2){
        //为了保证两条流的类型不一致，对第一条流进行修改
        DataStream<String> firstStream = sensorDataStream1.map(
            sensor -> sensor.getSensorId()+"\t"+sensor.getTemperature()
        );

        //两条不同类型的流进行合并
        ConnectedStreams<String,Demo02_sensorReading> connectedStream = firstStream.connect(sensorDataStream2);

        DataStream<Tuple2<String,Double>> resultDataStream = connectedStream.map(
            new CoMapFunction<String, Demo02_sensorReading, Tuple2<String,Double>>() {
                @Override
                public Tuple2<String, Double> map1(String value) throws Exception {
                    String[] fields = value.split("\t");
                    return Tuple2.of(fields[0],Double.parseDouble(fields[1]));
                }

                @Override
                public Tuple2<String, Double> map2(Demo02_sensorReading value) throws Exception {
                    return Tuple2.of(value.getSensorId(),value.getTemperature());
                }
            }
        );

        resultDataStream.print();
    }


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //接收原始数据
        DataStreamSource<String> dataStream1 = Demo02_source.source_socket(env,9999);
        DataStreamSource<String> dataStream2 = Demo02_source.source_socket(env,8888);

        DataStream<Demo02_sensorReading> sensorDataStream1 = transformation_simple(dataStream1);
        DataStream<Demo02_sensorReading> sensorDataStream2 = transformation_simple(dataStream2);

//        transformation_keyBy(sensorDataStream);
//        transformation_split(sensorDataStream);

//        transformation_union(sensorDataStream1,sensorDataStream2);

        transformation_connect(sensorDataStream1,sensorDataStream2);



        try {
            env.execute("transformation");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
